package spring.cloud.tasks.tasks_manager.task_item_assign;

import com.alibaba.fastjson2.JSON;
import com.google.common.cache.LoadingCache;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.api.transaction.CuratorTransaction;
import org.apache.curator.framework.api.transaction.CuratorTransactionFinal;
import org.apache.curator.framework.api.transaction.CuratorTransactionResult;
import spring.cloud.tasks.common.*;
import spring.cloud.tasks.common.curator.Curator;
import spring.cloud.tasks.common.nodepaths.*;
import spring.cloud.tasks.common.status.TaskInstanceInfoStatus;
import spring.cloud.tasks.common.status.TaskStageStatus;
import spring.cloud.tasks.common.task_item_allot.Executor;
import spring.cloud.tasks.common.task_item_allot.TaskItem;
import spring.cloud.tasks.tasks_manager.domain.WaitedAssignTaskInfo;
import spring.cloud.tasks.tasks_manager.schedule.ScheduleManager;
import spring.cloud.tasks.tasks_manager.schedule.TaskInstanceInfo;
import spring.cloud.tasks.tasks_manager.waitingPreAssignTasks.SlidingTimeTaskBucketQueueService;

import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.stream.Collectors;

@Slf4j
@Setter
@Getter
public class TaskItemAssignUtils {
    /**
     * 获取该命名空间下的所有task id list。
     * 整个命名空间,需要控制总的任务数据(实例信息+分片信息+续期信息)，不能超过zk传输的最大包。
     */
    public static List<String> getTaskIdList(Curator curator) throws Exception {
        String tasksNodePath = TaskNodePaths.getROOT();
        if (!curator.checkExists(tasksNodePath)) {
            return new ArrayList<>();
        }
        List<String> taskIdList = curator.getChildren(tasksNodePath);
        if (taskIdList != null) {
            return taskIdList;
        } else {
            return new ArrayList<>();
        }
    }

    /**
     * 获取该命名空间下的所有enable task id list。
     */
    public static List<String> getEnableTaskIdList(Curator curator, List<String> taskIdList) throws Exception {
        String enableTasksNodePath = EnabledTaskNodePaths.getRoot();
        if (!curator.checkExists(enableTasksNodePath)) {
            return new ArrayList<>();
        }
        List<String> enableTaskIdList = curator.getChildren(enableTasksNodePath);
        if (enableTaskIdList != null) {
            enableTaskIdList.removeAll(taskIdList);
            return enableTaskIdList;
        } else {
            return new ArrayList<>();
        }
    }

    /**
     * 获取任务列表+启用+暂停任务的列表,不新产生实例，但是老的实例还是会继续运行
     */
    public static List<String> getPauseTaskIdList(Curator curator) throws Exception {
        String pausedTaskNodePath = PausedTaskNodePaths.getRoot();
        if (!curator.checkExists(pausedTaskNodePath)) {
            return new ArrayList<>();
        }
        List<String> pausedTaskIdList = curator.getChildren(pausedTaskNodePath);
        if (pausedTaskIdList != null) {
            return pausedTaskIdList;
        } else {
            return new ArrayList<>();
        }
    }

    /**
     * 获取该命名空间下获取最新的执行器列表。
     */
    public static List<Executor> getLatestExecutorList(Curator curator) throws Exception {
        String executorsNodePath = ExecutorNodePaths.getRoot();
        if (!(curator.checkExists(executorsNodePath))) {
            return new ArrayList<>();
        }
        //需要控制总的任务数据，不能超过zk传输的最大包。
        List<String> executorIdList = curator.getChildren(ExecutorNodePaths.getExecutorSubNodePath(executorsNodePath));
        if (executorIdList == null) {
            return new ArrayList<>();
        }
        /**
         * 获取版本信息+获取ip信息。
         * 如果执行器节点数在1024内，那么是在220毫秒内(一类节点读取数据操作)。
         */
        //获取版本信息
        Map<String, Long> executorIdToVersionMap = new HashMap<>();
        for (String executorId : executorIdList) {
            String version = getVersion(curator, executorId);
            if (version == null) {
                continue;
            } else {
                executorIdToVersionMap.put(executorId, Long.valueOf(version));
            }
        }
        //获取ip信息
        Map<String, String> executorIdToIpMap = new HashMap<>();
        for (String executorId : executorIdList) {
            String ip = getIp(curator, executorId);
            if (ip == null) {
                continue;
            } else {
                executorIdToIpMap.put(executorId, ip);
            }
        }
        //
        String executorFlowControlNodePath = ExecutorFlowControlNodePaths.getRoot();
        List<String> executorFlowControlExecutorIdList = curator.getChildren(executorFlowControlNodePath);
        if (executorFlowControlExecutorIdList == null) {
            executorFlowControlExecutorIdList = new ArrayList<>();
        }
        //
        List<Executor> executorList = new ArrayList<>();
        for (int i = 0; i < executorIdList.size(); i++) {
            String executorId = executorIdList.get(i);
            Long version = executorIdToVersionMap.get(executorId);
            if (version == null) {
                continue;//没有版本信息，则代表该节点不存在
            }
            String ip = executorIdToIpMap.get(executorId);
            if (ip == null) {//ip是一个瞬时节点,有值代表在线,无值代表还没有上线完毕
                continue;
            }
            //
            Executor executor = new Executor();
            executor.setExecutorId(executorId);
            executor.setIp(ip);
            executor.setNotFlowControl(!executorFlowControlExecutorIdList.contains(executorId));
            executor.setTaskIdList(getTaskIdList(curator, executorId, version));
            executor.setTaskItemList(new ArrayList<>());//init
            executorList.add(executor);
        }
        return executorList;
    }

    /**
     * 获取执行器的版本
     */
    public static String getVersion(Curator curator, String executorId) throws Exception {
        String version = null;
        String executorNodePath = ExecutorNodePaths.getExecutorSubNodePath(executorId);
        if (curator.checkExists(ExecutorNodePaths.getExecutorIpNodePath(executorId))) {
            version = curator.getData(executorNodePath);
        }
        return version;
    }

    /**
     * 获取执行器的ip
     */
    public static String getIp(Curator curator, String executorId) throws Exception {
        String ip = null;
        String executorIpNodePath = ExecutorNodePaths.getExecutorIpNodePath(executorId);
        if (curator.checkExists(ExecutorNodePaths.getExecutorIpNodePath(executorId))) {
            ip = curator.getData(executorIpNodePath);
        }
        return ip;
    }

    /**
     * 获取执行器是否被流控
     */
    public static boolean executorIsNotFlowControl(Curator curator, String executorId) throws Exception {
        String executorFlowControlNodePath = ExecutorFlowControlNodePaths.getExecutorFlowControlNodePath(executorId);
        return !curator.checkExists(executorFlowControlNodePath);

    }

    public static List<String> getTaskIdList(Curator curator, String executorId, long version) throws Exception {
        LoadingCache<String, ExecutorSupportedTaskIdList> cache = TaskItemContainer.executorIdToExecutorSupportedTaskIdListCache;
        ExecutorSupportedTaskIdList executorSupportedTaskIdList = cache.get(executorId);
        if (executorSupportedTaskIdList != null && executorSupportedTaskIdList != ExecutorSupportedTaskIdList.$null$) {
            if (executorSupportedTaskIdList.version == version) {
                return executorSupportedTaskIdList.getTaskIdList();
            }
        }
        List<String> taskIdList = new ArrayList<>();
        String executorTaskIdListNodePath = ExecutorNodePaths.getExecutorSubNodePath(executorId, "taskIdList");
        if (curator.checkExists(executorTaskIdListNodePath)) {
            taskIdList = curator.getChildren(executorTaskIdListNodePath);
        }
        executorSupportedTaskIdList = new ExecutorSupportedTaskIdList();
        executorSupportedTaskIdList.setExecutorId(executorId);
        executorSupportedTaskIdList.setVersion(version);
        executorSupportedTaskIdList.setTaskIdList(taskIdList);
        cache.put(executorId, executorSupportedTaskIdList);
        return taskIdList;
    }

    public static List<String> getExecutorIdList(List<Executor> executorList) {
        List<String> list = new ArrayList<>();
        for (Executor executor : executorList) {
            list.add(executor.getExecutorId());
        }
        return list;
    }

    /**
     * 获取没有被流控的执行器，返回的是引用，不是复制。
     */
    public static List<Executor> getIsNotFlowControlExecutorList(List<Executor> executorList) {
        List<Executor> list = new ArrayList<>();
        for (Executor executor : executorList) {
            if (executor.isNotFlowControl()) {
                list.add(executor);
            }
        }
        return list;
    }

    public static List<String> getIsNotFlowControlExecutorIdList(List<Executor> executorList) {
        List<String> list = new ArrayList<>();
        for (Executor executor : executorList) {
            if (executor.isNotFlowControl()) {
                list.add(executor.getExecutorId());
            }
        }
        return list;
    }

    //
    public static List<Executor> getFlowControlExecutorList(List<Executor> executorList) {
        List<Executor> list = new ArrayList<>();
        for (Executor executor : executorList) {
            if (!executor.isNotFlowControl()) {
                list.add(executor);
            }
        }
        return list;
    }

    public static List<String> getFlowControlExecutorIdList(List<Executor> executorList) {
        List<String> list = new ArrayList<>();
        for (Executor executor : executorList) {
            if (!executor.isNotFlowControl()) {
                list.add(executor.getExecutorId());
            }
        }
        return list;
    }

    public static List<Executor> copyExecutorList(List<Executor> executorList) {
        List<Executor> list = new ArrayList<>();
        for (Executor executor : executorList) {
            Executor executorCopy = new Executor();
            executorCopy.setExecutorId(executor.getExecutorId());
            executorCopy.setIp(executor.getIp());
            executorCopy.setNotFlowControl(executor.isNotFlowControl());
            executorCopy.setTotalLoadLevel(executor.getTotalLoadLevel());
            executorCopy.getTaskIdList().addAll(executor.getTaskIdList());//[ArrayList.addAll元素拷贝+原始就是值对象=值对象元素拷贝]
            for (TaskItem taskItem : executor.getTaskItemList()) {
                TaskItem taskItemCopy = new TaskItem();
                taskItemCopy.copy(taskItem);
                executorCopy.getTaskItemList().add(taskItemCopy);
            }
            list.add(executorCopy);
        }
        return list;
    }

    //##<!-我是优美的功能模块分割线->##
    public static List<TaskItem> getWaitedAssignTaskItemList(//
                                                             ScheduleManager scheduleManager,
                                                             Curator curator,//
                                                             SlidingTimeTaskBucketQueueService slidingTimeTaskBucketQueueService,
                                                             List<String> enableTaskIdList,
                                                             List<TaskInstanceInfo> waitedAssignTaskInstanceInfoList
    ) throws Exception {//

        List<TaskInstanceInfo> taskInstanceInfoList = slidingTimeTaskBucketQueueService.getTaskInstanceInfoList();
        for (TaskInstanceInfo taskInstanceInfo : taskInstanceInfoList) {
            String taskType = taskInstanceInfo.getTaskType();
            String taskId = taskInstanceInfo.getTaskId();
            if (!enableTaskIdList.contains(taskId)) {
                taskInstanceInfo.setCancelled(true);
                continue;
            }
            if (TaskType.SIMPLE_SLICING.equals(taskType)) {
                String stateFullTask = taskInstanceInfo.getStateFullTask();
                if (YN.Y.equals(stateFullTask)) {
                    scheduleManager.stateFullSimpleTaskTaskInstanceManager.waitingAssignTasks.put(taskId, new CopyOnWriteArrayList<>());
                    scheduleManager.stateFullSimpleTaskTaskInstanceManager.waitingAssignTasks.get(taskId).add(taskInstanceInfo);
                } else {
                    scheduleManager.stateLessSimpleTaskTaskInstanceManager.waitingAssignTasks.put(taskId, new CopyOnWriteArrayList<>());
                    scheduleManager.stateLessSimpleTaskTaskInstanceManager.waitingAssignTasks.get(taskId).add(taskInstanceInfo);
                }
            } else if (TaskType.TIMING_BASED_BATCH_PROCESSING.getTaskType().equals(taskType) || TaskType.USER_TIMING_BASED_BATCH_PROCESSING.getTaskType().equals(taskType)) {
                scheduleManager.batchTaskTaskInstanceManager.waitingAssignTasks.put(taskId, new CopyOnWriteArrayList<>());
                scheduleManager.batchTaskTaskInstanceManager.waitingAssignTasks.get(taskId).add(taskInstanceInfo);
            } else {
                throw new IllegalStateException();
            }
        }
        //先进行上面的实例进行状态改变为预调度-尝试预调度。（尝试预调度  预调度中  预调度完成）
        List<Collection<CopyOnWriteArrayList<TaskInstanceInfo>>> collectionList = new ArrayList<>();
        Collection<CopyOnWriteArrayList<TaskInstanceInfo>> values1 = scheduleManager.batchTaskTaskInstanceManager.waitingAssignTasks.values();
        Collection<CopyOnWriteArrayList<TaskInstanceInfo>> values2 = scheduleManager.stateFullSimpleTaskTaskInstanceManager.waitingAssignTasks.values();
        Collection<CopyOnWriteArrayList<TaskInstanceInfo>> values3 = scheduleManager.stateLessSimpleTaskTaskInstanceManager.waitingAssignTasks.values();
        collectionList.add(values1);
        collectionList.add(values2);
        collectionList.add(values3);
        //
        List<TaskInstanceInfo> list = new ArrayList<>();
        CuratorTransaction curatorTransaction = curator.getCuratorFramework().inTransaction();
        CuratorTransactionFinal curatorTransactionFinal = (CuratorTransactionFinal) curatorTransaction;
        for (Collection<CopyOnWriteArrayList<TaskInstanceInfo>> collection : collectionList) {
            for (List<TaskInstanceInfo> e : collection) {
                for (TaskInstanceInfo taskInstanceInfo : e) {
                    if (taskInstanceInfo.isCancelled()) {
                        continue;//如果任务取消了但是还没有添加到预分配列表里面。那么就不添加了。
                    }
                    if (!enableTaskIdList.contains(taskInstanceInfo.getTaskId())) {
                        taskInstanceInfo.setCancelled(true);
                        continue;//打上取消标志
                    }
                    list.add(taskInstanceInfo);
                    String taskInstanceStatusNodePath = TaskInstanceNodePaths.getTaskInstanceSubNodePath(taskInstanceInfo.getTaskId(), taskInstanceInfo.getTaskInstanceId(), TaskInstanceNodePaths.TaskInstanceSubNode.status);
                    curatorTransactionFinal = curatorTransactionFinal.setData().forPath(taskInstanceStatusNodePath, TaskInstanceInfoStatus.waiting_task_item_pre_assign_success.getBytes(StandardCharsets.UTF_8)).and();
                }
            }

        }
        Collection<CuratorTransactionResult> commit = curatorTransactionFinal.commit();
        for (CuratorTransactionResult curatorTransactionResult : commit) {
            System.out.println(curatorTransactionResult);
        }
        //上述的实例信息进行预分配后 那么就需要被注册到ZK。当前调度器挂了以后任然可以进行恢复。
        //先进行实例信息待调度的预写  WAL机制。
        List<TaskItem> taskItemList = new ArrayList<>();
        for (TaskInstanceInfo taskInstanceInfo : list) {
            //暂时不进行取消
            //if (taskInstanceInfo.isCancelled()){ continue;}
            String taskItemInfo = taskInstanceInfo.getTaskItemInfo();
            TaskStageInfo taskStageAndTaskItemTypeAndCountAndLoadInfo = null;
            List<TaskStageInfo> taskStageAndTaskItemTypeAndCountAndLoadInfoList = JSON.parseArray(taskItemInfo, TaskStageInfo.class);
            for (TaskStageInfo stageAndTaskItemTypeAndCountAndLoadInfo : taskStageAndTaskItemTypeAndCountAndLoadInfoList) {
                if (stageAndTaskItemTypeAndCountAndLoadInfo.stage.equals(taskInstanceInfo.getCurrentStage())) {
                    taskStageAndTaskItemTypeAndCountAndLoadInfo = stageAndTaskItemTypeAndCountAndLoadInfo;
                    break;
                }
            }
            taskInstanceInfo.setCurrentTaskStageInfo(taskStageAndTaskItemTypeAndCountAndLoadInfo);
            taskItemList.addAll(buildBatchTaskNextTaskItemList(taskInstanceInfo));
            waitedAssignTaskInstanceInfoList.add(taskInstanceInfo);
        }
        return taskItemList;
    }

    public static void addWaitedAssignTaskItemList(ScheduleManager scheduleManager, List<TaskInstanceInfo> waitedAssignTaskInstanceInfoList) {
        for (TaskInstanceInfo taskInstanceInfo : waitedAssignTaskInstanceInfoList) {
            String taskId = taskInstanceInfo.getTaskId();
            String taskType = taskInstanceInfo.getTaskType();
            if (TaskType.SIMPLE_SLICING.equals(taskType)) {
                String stateFullTask = taskInstanceInfo.getStateFullTask();
                if (YN.Y.equals(stateFullTask)) {
                    scheduleManager.stateFullSimpleTaskTaskInstanceManager.assignedTasks.put(taskId, new CopyOnWriteArrayList<>());
                    scheduleManager.stateFullSimpleTaskTaskInstanceManager.assignedTasks.get(taskId).add(taskInstanceInfo);
                    //
                    scheduleManager.stateFullSimpleTaskTaskInstanceManager.waitingAssignTasks.put(taskId, new CopyOnWriteArrayList<>());
                    scheduleManager.stateFullSimpleTaskTaskInstanceManager.waitingAssignTasks.get(taskId).remove(taskInstanceInfo);
                } else {
                    scheduleManager.stateLessSimpleTaskTaskInstanceManager.assignedTasks.put(taskId, new CopyOnWriteArrayList<>());
                    scheduleManager.stateLessSimpleTaskTaskInstanceManager.assignedTasks.get(taskId).add(taskInstanceInfo);
                    //
                    scheduleManager.stateLessSimpleTaskTaskInstanceManager.waitingAssignTasks.put(taskId, new CopyOnWriteArrayList<>());
                    scheduleManager.stateLessSimpleTaskTaskInstanceManager.waitingAssignTasks.get(taskId).remove(taskInstanceInfo);
                }
            } else if (TaskType.TIMING_BASED_BATCH_PROCESSING.getTaskType().equals(taskType) || TaskType.USER_TIMING_BASED_BATCH_PROCESSING.getTaskType().equals(taskType)) {
                scheduleManager.batchTaskTaskInstanceManager.assignedTasks.put(taskId, new CopyOnWriteArrayList<>());
                scheduleManager.batchTaskTaskInstanceManager.assignedTasks.get(taskId).add(taskInstanceInfo);
                //
                scheduleManager.batchTaskTaskInstanceManager.waitingAssignTasks.put(taskId, new CopyOnWriteArrayList<>());
                scheduleManager.batchTaskTaskInstanceManager.waitingAssignTasks.get(taskId).remove(taskInstanceInfo);
            } else {
                throw new IllegalStateException();
            }
        }
    }

    public static TaskInstanceInfo getTaskInstanceInfo(ScheduleManager scheduleManager, String taskId, String taskType, String stateFullTask, String taskInstanceId) {
        if (TaskType.SIMPLE_SLICING.getTaskType().equals(taskType) && YN.Y.equals(stateFullTask)) {

            {
                List<TaskInstanceInfo> taskInstanceInfoList = scheduleManager.stateFullSimpleTaskTaskInstanceManager.assignedTasks.get(taskId);
                if (taskInstanceInfoList != null) {
                    for (TaskInstanceInfo taskInstanceInfo : taskInstanceInfoList) {
                        if (taskInstanceInfo.getTaskInstanceId().equals(taskInstanceId)) {
                            return taskInstanceInfo;
                        }
                    }
                }
            }

            {
                List<TaskInstanceInfo> taskInstanceInfoList = scheduleManager.stateFullSimpleTaskTaskInstanceManager.waitingAssignTasks.get(taskId);
                if (taskInstanceInfoList != null) {
                    for (TaskInstanceInfo taskInstanceInfo : taskInstanceInfoList) {
                        if (taskInstanceInfo.getTaskInstanceId().equals(taskInstanceId)) {
                            return taskInstanceInfo;
                        }
                    }
                }
            }
            return null;
        } else if (TaskType.SIMPLE_SLICING.getTaskType().equals(taskType) && YN.Y.equals(stateFullTask)) {
            {
                List<TaskInstanceInfo> taskInstanceInfoList = scheduleManager.stateLessSimpleTaskTaskInstanceManager.assignedTasks.get(taskId);
                if (taskInstanceInfoList != null) {
                    for (TaskInstanceInfo taskInstanceInfo : taskInstanceInfoList) {
                        if (taskInstanceInfo.getTaskInstanceId().equals(taskInstanceId)) {
                            if (taskInstanceInfo.getTaskInstanceId().equals(taskInstanceId)) {
                                return taskInstanceInfo;
                            }
                        }
                    }
                }
            }
            {
                List<TaskInstanceInfo> taskInstanceInfoList = scheduleManager.stateLessSimpleTaskTaskInstanceManager.waitingAssignTasks.get(taskId);
                if (taskInstanceInfoList != null) {
                    for (TaskInstanceInfo taskInstanceInfo : taskInstanceInfoList) {
                        if (taskInstanceInfo.getTaskInstanceId().equals(taskInstanceId)) {
                            if (taskInstanceInfo.getTaskInstanceId().equals(taskInstanceId)) {
                                return taskInstanceInfo;
                            }
                        }
                    }
                }
            }
            return null;
        } else if (TaskType.TIMING_BASED_BATCH_PROCESSING.getTaskType().equals(taskType) || TaskType.USER_TIMING_BASED_BATCH_PROCESSING.getTaskType().equals(taskType)) {
            {
                List<TaskInstanceInfo> taskInstanceInfoList = scheduleManager.batchTaskTaskInstanceManager.assignedTasks.get(taskId);
                if (taskInstanceInfoList != null) {
                    for (TaskInstanceInfo taskInstanceInfo : taskInstanceInfoList) {
                        if (taskInstanceInfo.getTaskInstanceId().equals(taskInstanceId)) {
                            if (taskInstanceInfo.getTaskInstanceId().equals(taskInstanceId)) {
                                return taskInstanceInfo;
                            }
                        }
                    }
                }
            }
            {
                List<TaskInstanceInfo> taskInstanceInfoList = scheduleManager.batchTaskTaskInstanceManager.waitingAssignTasks.get(taskId);
                if (taskInstanceInfoList != null) {
                    for (TaskInstanceInfo taskInstanceInfo : taskInstanceInfoList) {
                        if (taskInstanceInfo.getTaskInstanceId().equals(taskInstanceId)) {
                            if (taskInstanceInfo.getTaskInstanceId().equals(taskInstanceId)) {
                                return taskInstanceInfo;
                            }
                        }
                    }
                }
            }
            return null;
        } else {
            throw new IllegalStateException();
        }


    }

    public static List<TaskInstanceInfo> getTaskInstanceInfoList(ScheduleManager scheduleManager, String taskId, String taskType, String stateFullTask) {
        List<TaskInstanceInfo> list = new ArrayList<>();
        if (TaskType.SIMPLE_SLICING.getTaskType().equals(taskType) && YN.Y.equals(stateFullTask)) {

            {
                List<TaskInstanceInfo> taskInstanceInfoList = scheduleManager.stateFullSimpleTaskTaskInstanceManager.assignedTasks.get(taskId);
                if (taskInstanceInfoList != null) {
                    for (TaskInstanceInfo taskInstanceInfo : taskInstanceInfoList) {
                        list.add(taskInstanceInfo);
                    }
                }
            }

            {
                List<TaskInstanceInfo> taskInstanceInfoList = scheduleManager.stateFullSimpleTaskTaskInstanceManager.waitingAssignTasks.get(taskId);
                if (taskInstanceInfoList != null) {
                    for (TaskInstanceInfo taskInstanceInfo : taskInstanceInfoList) {
                        list.add(taskInstanceInfo);
                    }
                }
            }
            return list;
        } else if (TaskType.SIMPLE_SLICING.getTaskType().equals(taskType) && YN.Y.equals(stateFullTask)) {
            {
                List<TaskInstanceInfo> taskInstanceInfoList = scheduleManager.stateLessSimpleTaskTaskInstanceManager.assignedTasks.get(taskId);
                if (taskInstanceInfoList != null) {
                    for (TaskInstanceInfo taskInstanceInfo : taskInstanceInfoList) {
                        list.add(taskInstanceInfo);
                    }
                }
            }
            {
                List<TaskInstanceInfo> taskInstanceInfoList = scheduleManager.stateLessSimpleTaskTaskInstanceManager.waitingAssignTasks.get(taskId);
                if (taskInstanceInfoList != null) {
                    for (TaskInstanceInfo taskInstanceInfo : taskInstanceInfoList) {
                        list.add(taskInstanceInfo);
                    }
                }
            }
            return list;
        } else if (TaskType.TIMING_BASED_BATCH_PROCESSING.getTaskType().equals(taskType) || TaskType.USER_TIMING_BASED_BATCH_PROCESSING.getTaskType().equals(taskType)) {
            {
                List<TaskInstanceInfo> taskInstanceInfoList = scheduleManager.batchTaskTaskInstanceManager.assignedTasks.get(taskId);
                if (taskInstanceInfoList != null) {
                    for (TaskInstanceInfo taskInstanceInfo : taskInstanceInfoList) {
                        list.add(taskInstanceInfo);
                    }
                }
            }
            {
                List<TaskInstanceInfo> taskInstanceInfoList = scheduleManager.batchTaskTaskInstanceManager.waitingAssignTasks.get(taskId);
                if (taskInstanceInfoList != null) {
                    for (TaskInstanceInfo taskInstanceInfo : taskInstanceInfoList) {
                        list.add(taskInstanceInfo);
                    }
                }
            }
            return list;
        } else {
            throw new IllegalStateException();
        }


    }

    public static void updateInstanceStatusWhenUserCancel(ScheduleManager scheduleManager, String taskId, String taskInstanceId, String status) {

        {
            List<TaskInstanceInfo> taskInstanceInfoList = scheduleManager.stateFullSimpleTaskTaskInstanceManager.assignedTasks.get(taskId);
            if (taskInstanceInfoList != null) {
                for (TaskInstanceInfo taskInstanceInfo : taskInstanceInfoList) {
                    if (taskInstanceInfo.getTaskInstanceId().equals(taskInstanceId)) {
                        taskInstanceInfo.setStatus(status);
                        return;
                    }
                }
            }
        }
        {
            List<TaskInstanceInfo> taskInstanceInfoList = scheduleManager.stateLessSimpleTaskTaskInstanceManager.assignedTasks.get(taskId);
            if (taskInstanceInfoList != null) {
                for (TaskInstanceInfo taskInstanceInfo : taskInstanceInfoList) {
                    if (taskInstanceInfo.getTaskInstanceId().equals(taskInstanceId)) {
                        taskInstanceInfo.setStatus(status);
                        return;
                    }
                }
            }
        }
        {
            List<TaskInstanceInfo> taskInstanceInfoList = scheduleManager.batchTaskTaskInstanceManager.assignedTasks.get(taskId);
            if (taskInstanceInfoList != null) {
                for (TaskInstanceInfo taskInstanceInfo : taskInstanceInfoList) {
                    if (taskInstanceInfo.getTaskInstanceId().equals(taskInstanceId)) {
                        taskInstanceInfo.setStatus(status);
                        return;
                    }
                }
            }
        }
    }

    /**
     * 这个是底层方法
     */
    public static List<TaskItem> buildBatchTaskNextTaskItemList(TaskInstanceInfo taskInstanceInfo) {
        if (taskInstanceInfo.getTaskType().equals(TaskType.SIMPLE_SLICING)) {
            String taskItemInfo = taskInstanceInfo.getTaskItemInfo();
            SimpleSlicingTaskTaskItemCountAndLoadAndParameterInfo simpleSlicingTaskTaskItemCountAndLoadAndParameterInfo = JSON.parseObject(taskItemInfo, SimpleSlicingTaskTaskItemCountAndLoadAndParameterInfo.class);
            //创建所有分片
            int taskItemCount = simpleSlicingTaskTaskItemCountAndLoadAndParameterInfo.taskItemCount;
            int taskItemLoad = simpleSlicingTaskTaskItemCountAndLoadAndParameterInfo.taskItemLoad;
            String taskItemIndexAndTaskItemParametersString = simpleSlicingTaskTaskItemCountAndLoadAndParameterInfo.taskItemIndexAndTaskItemParametersString;
            String taskItemPropertiesString = simpleSlicingTaskTaskItemCountAndLoadAndParameterInfo.taskItemPropertiesString;
            WaitedAssignTaskInfo waitedAssignTaskInfo = new WaitedAssignTaskInfo();
            String taskId = taskInstanceInfo.getTaskId();
            waitedAssignTaskInfo.setTaskId(taskId);
            String taskType = taskInstanceInfo.getTaskType();
            waitedAssignTaskInfo.setTaskType(taskType);
            String taskInstanceId = taskInstanceInfo.getTaskInstanceId();
            waitedAssignTaskInfo.setTaskInstanceId(taskInstanceId);
            waitedAssignTaskInfo.setTaskStage(TaskType.SimpleSlicingTask.stage_one);
            waitedAssignTaskInfo.setTaskStage$type((YN.Y.equals(taskInstanceInfo.getStateFullTask()) ? TaskType.SimpleSlicingTask.stage_type_state_full : TaskType.SimpleSlicingTask.stage_type_state_less));
            waitedAssignTaskInfo.setTaskItemCount(taskItemCount);
            waitedAssignTaskInfo.setTaskItemLoad(taskItemLoad);
            Date nextFireDate = taskInstanceInfo.getNextFireDate();
            waitedAssignTaskInfo.setNextFireDate(nextFireDate);
            Date timeoutDateTime = new Date(nextFireDate.getTime() + 1000L * 60 * taskInstanceInfo.getTimeoutMinutes());//超时时间
            waitedAssignTaskInfo.setTimeoutDateTime(timeoutDateTime);
            Date lockDateTime = new Date(timeoutDateTime.getTime() + 1000L * 60 * taskInstanceInfo.getLockMinutes());//lock时间
            waitedAssignTaskInfo.setLockDateTime(lockDateTime);
            List<TaskItem> taskItemList = createTaskItemList(waitedAssignTaskInfo);
            if (taskInstanceInfo.isTryCancel()) {
                taskInstanceInfo.setStatus(TaskInstanceInfoStatus.user_cancel);
                taskInstanceInfo.setTaskItemList(new ArrayList<>());
                return new ArrayList<>();
            }
            return taskItemList;
            /**
             * init构建和重跑构建，都是基于处于某个阶段进行创建对应的分片。重跑可以指定数据范围和执行的阶段.这个属于实例重新跑，但是不一定是从实例的第一个阶段跑。
             */
        } else if (taskInstanceInfo.getTaskType().equals(TaskType.TIMING_BASED_BATCH_PROCESSING) || taskInstanceInfo.getTaskType().equals(TaskType.USER_TIMING_BASED_BATCH_PROCESSING)) {
            if (taskInstanceInfo.isCancelled()) {
                taskInstanceInfo.setStatus(TaskInstanceInfoStatus.user_cancel);
                taskInstanceInfo.setTaskItemList(new ArrayList<>());
                taskInstanceInfo.setRebuildAssignTaskItemPlanCreate(false);
                return new ArrayList<>();
            }
            String currentStage = taskInstanceInfo.getCurrentStage();
            String currentStageStatus = taskInstanceInfo.getCurrentStageStatus();
            if (!TaskStageStatus.status_watting_init.equals(currentStageStatus)) {//
                log.warn("重跑计算任务实例信息的状态必须是init状态");
                taskInstanceInfo.setStatus(TaskInstanceInfoStatus.user_cancel);
                taskInstanceInfo.setTaskItemList(new ArrayList<>());
                taskInstanceInfo.setRebuildAssignTaskItemPlanCreate(false);
                return new ArrayList<>();
            }
            String taskItemInfo = taskInstanceInfo.getTaskItemInfo();
            List<TaskStageInfo> taskStageAndTaskItemTypeAndCountAndLoadInfoList = JSON.parseArray(taskItemInfo, TaskStageInfo.class);
            Collections.sort(taskStageAndTaskItemTypeAndCountAndLoadInfoList, new Comparator<TaskStageInfo>() {
                @Override
                public int compare(TaskStageInfo o1, TaskStageInfo o2) {
                    return o1.stageNum - o2.stageNum;
                }
            });
            TaskStageInfo current = null;
            for (TaskStageInfo taskStageAndTaskItemTypeAndCountAndLoadInfo : taskStageAndTaskItemTypeAndCountAndLoadInfoList) {
                if (taskStageAndTaskItemTypeAndCountAndLoadInfo.stage.equals(currentStage)) {
                    current = taskStageAndTaskItemTypeAndCountAndLoadInfo;
                    taskInstanceInfo.setCurrentStage(taskStageAndTaskItemTypeAndCountAndLoadInfo.stage);
                    continue;
                }
            }
            if (current == null) {
                //bug fix
                taskInstanceInfo.setStatus(TaskInstanceInfoStatus.status_stage_not_find);
                taskInstanceInfo.setTaskItemList(new ArrayList<>());
                taskInstanceInfo.setRebuildAssignTaskItemPlanCreate(false);
                return new ArrayList<>();
            }
            taskInstanceInfo.setCurrentTaskStageInfo(current);
            Date nextFireDate = new Date();
            Date timeoutDateTime = new Date(nextFireDate.getTime() + DateUtils.MINUTE * taskInstanceInfo.getTimeoutMinutes());
            Date lockDateTime = new Date(timeoutDateTime.getTime() + DateUtils.MINUTE * taskInstanceInfo.getLockMinutes());
            Date clearDateTime = new Date(lockDateTime.getTime() + DateUtils.MINUTE * taskInstanceInfo.getClearMinutes());
            taskInstanceInfo.setNextFireDate(nextFireDate);
            taskInstanceInfo.setTimeoutDateTime(timeoutDateTime);
            taskInstanceInfo.setLockDateTime(lockDateTime);
            taskInstanceInfo.setClearDateTime(clearDateTime);
            List<TaskStageInfo.TaskItemInfo> taskItemTypeAndCountAndLoadList = current.taskItemInfoList;
            List<TaskItem> taskItemList = new ArrayList<>();
            for (TaskStageInfo.TaskItemInfo taskItemTypeAndCountAndLoad : taskItemTypeAndCountAndLoadList) {
                WaitedAssignTaskInfo waitedAssignTaskInfo = new WaitedAssignTaskInfo();
                String taskId = taskInstanceInfo.getTaskId();
                waitedAssignTaskInfo.setTaskId(taskId);
                String taskType = taskInstanceInfo.getTaskType();
                waitedAssignTaskInfo.setTaskType(taskType);
                String taskInstanceId = taskInstanceInfo.getTaskInstanceId();
                waitedAssignTaskInfo.setTaskInstanceId(taskInstanceId);
                waitedAssignTaskInfo.setTaskStage(current.stage);
                waitedAssignTaskInfo.setTaskStage$type(taskItemTypeAndCountAndLoad.taskItemType);
                int taskItemCount = taskItemTypeAndCountAndLoad.taskItemCount;
                int taskItemLoad = taskItemTypeAndCountAndLoad.taskItemLoad;
                waitedAssignTaskInfo.setTaskItemCount(taskItemCount);
                waitedAssignTaskInfo.setTaskItemLoad(taskItemLoad);
                //Date nextFireDate = taskInstanceInfo.getNextFireDate();
                waitedAssignTaskInfo.setNextFireDate(nextFireDate);
                //Date timeoutDateTime = new Date(nextFireDate.getTime() + DateUtils.MINUTE * taskInstanceInfo.getTimeoutMinutes());//超时时间
                waitedAssignTaskInfo.setTimeoutDateTime(timeoutDateTime);
                //Date lockDateTime = new Date(timeoutDateTime.getTime() + DateUtils.MINUTE * taskInstanceInfo.getLockMinutes());//lock时间
                waitedAssignTaskInfo.setLockDateTime(lockDateTime);
                //Date clearDateTime = new Date(lockDateTime.getTime() + DateUtils.MINUTE * taskInstanceInfo.getClearMinutes());
                waitedAssignTaskInfo.setClearDateTime(clearDateTime);
                List<TaskItem> $taskItemList$ = createTaskItemList(waitedAssignTaskInfo);
                taskItemList.addAll($taskItemList$);
            }
            taskInstanceInfo.setRebuildAssignTaskItemPlanCreate(true);
            taskInstanceInfo.setCurrentStageStatus(TaskStageStatus.status_handing);
            if (taskInstanceInfo.isCancelled()) {
                taskInstanceInfo.setStatus(TaskInstanceInfoStatus.user_cancel);
                taskInstanceInfo.setTaskItemList(new ArrayList<>());
                taskInstanceInfo.setRebuildAssignTaskItemPlanCreate(false);
                return new ArrayList<>();
            }
            return taskItemList;
        } else {
            throw new UnsupportedOperationException();
        }

    }

    /**
     * 正常跑构建
     *
     * @param taskInstanceInfo
     */
    public static void rebuildBatchTaskNextTaskItemList(TaskInstanceInfo taskInstanceInfo) {
        if (taskInstanceInfo.isCancelled()) {
            taskInstanceInfo.setStatus(TaskInstanceInfoStatus.user_cancel);
            taskInstanceInfo.setTaskItemList(new ArrayList<>());
            taskInstanceInfo.setRebuildAssignTaskItemPlanCreate(false);
            return;
        }
        String currentStage = taskInstanceInfo.getCurrentStage();
        String currentStageStatus = taskInstanceInfo.getCurrentStageStatus();
        if (!TaskStageStatus.status_watting_init.equals(currentStageStatus)) {//
            return;
        }
        /**
         * 只要没有到超时时间，该实例当前阶段的分片列表，可以无限次重新预分配，直到成功，不会产生副作用。
         */
        if (TaskStageStatus.status_handing.equals(currentStageStatus)) {//
            //计算是否超时
            //标记当前任务实例某个stage超时,实例整体超时，后续的stage不能执行。可以允许重启进行执行。
            //如果超时了那么当前stage停止执行。整个任务状态标记为超时。
            if (taskInstanceInfo.isTimeout()) {
                taskInstanceInfo.setStatus(TaskInstanceInfoStatus.status_timeout);
                taskInstanceInfo.setRebuildAssignTaskItemPlanCreate(false);
                return;
            } else {
                //...
                return;
            }//什么时候taskStage会变成处理成功或者处理失败呢。需要状态通知节点进行主动通知。比如捞取数据进行数据拆分节点，就会主动通知。以及后续的专门的通知节点进行通知。
        }
        if (TaskStageStatus.status_success.equals(currentStageStatus)) {//
            //进行下一个stage的taskItem构建
            String taskItemInfo = taskInstanceInfo.getTaskItemInfo();
            List<TaskStageInfo> taskStageAndTaskItemTypeAndCountAndLoadInfoList = JSON.parseArray(taskItemInfo, TaskStageInfo.class);
            Collections.sort(taskStageAndTaskItemTypeAndCountAndLoadInfoList, new Comparator<TaskStageInfo>() {
                @Override
                public int compare(TaskStageInfo o1, TaskStageInfo o2) {
                    return o1.stageNum - o2.stageNum;
                }
            });
            int size = taskStageAndTaskItemTypeAndCountAndLoadInfoList.size();
            Map<String, Integer> stageToStageNumMap = new HashMap<>();
            Map<String, String> stageNumToStageMap = new HashMap<>();
            for (TaskStageInfo taskStageAndTaskItemTypeAndCountAndLoadInfo : taskStageAndTaskItemTypeAndCountAndLoadInfoList) {
                String stage = taskStageAndTaskItemTypeAndCountAndLoadInfo.stage;
                int stageNum = taskStageAndTaskItemTypeAndCountAndLoadInfo.stageNum;
                stageToStageNumMap.put(stage, stageNum);
                stageNumToStageMap.put(String.valueOf(stageNum), stage);
            }
            int stageNum = stageToStageNumMap.get(currentStage);
            if (size == stageNum) {
                taskInstanceInfo.setStatus(TaskInstanceInfoStatus.status_success);//可以吧当前的taskItem进行待清理
                taskInstanceInfo.setRebuildAssignTaskItemPlanCreate(false);
                return;
            }
            //否则生成下个阶段的执行计划。
            String nextStage = stageNumToStageMap.get(String.valueOf(stageNum + 1));
            TaskStageInfo next = null;
            for (TaskStageInfo taskStageAndTaskItemTypeAndCountAndLoadInfo : taskStageAndTaskItemTypeAndCountAndLoadInfoList) {
                if (taskStageAndTaskItemTypeAndCountAndLoadInfo.stage.equals(nextStage)) {
                    next = taskStageAndTaskItemTypeAndCountAndLoadInfo;
                    break;
                }
            }
            if (next == null) {
                //bug fix
                taskInstanceInfo.setStatus(TaskInstanceInfoStatus.status_stage_not_find);
                taskInstanceInfo.setTaskItemList(new ArrayList<>());
                taskInstanceInfo.setRebuildAssignTaskItemPlanCreate(false);
                return;
            }
            taskInstanceInfo.setCurrentStage(nextStage);
            taskInstanceInfo.setCurrentTaskStageInfo(next);
            Date nextFireDate = new Date();
            Date timeoutDateTime = new Date(nextFireDate.getTime() + DateUtils.MINUTE * taskInstanceInfo.getTimeoutMinutes());
            Date lockDateTime = new Date(timeoutDateTime.getTime() + DateUtils.MINUTE * taskInstanceInfo.getLockMinutes());
            Date clearDateTime = new Date(lockDateTime.getTime() + DateUtils.MINUTE * taskInstanceInfo.getClearMinutes());
            taskInstanceInfo.setNextFireDate(nextFireDate);
            taskInstanceInfo.setTimeoutDateTime(timeoutDateTime);
            taskInstanceInfo.setLockDateTime(lockDateTime);
            taskInstanceInfo.setClearDateTime(clearDateTime);
            //
            List<TaskItem> taskItemList = TaskItemAssignUtils.buildBatchTaskNextTaskItemList(taskInstanceInfo);
            taskInstanceInfo.setTaskItemList(taskItemList);//需要清除当前实例非当前stage的任务分片
            taskInstanceInfo.setRebuildAssignTaskItemPlanCreate(true);
            //
            taskInstanceInfo.setCurrentStageStatus(TaskStageStatus.status_handing);
            return;

        }
        if (TaskStageStatus.status_fail.equals(currentStageStatus)) {//
            //标记当前任务实例某个stage失败,实例整体失败，后续的stage不能执行。可以允许重启进行执行。
            taskInstanceInfo.setStatus(TaskInstanceInfoStatus.status_fail);
            taskInstanceInfo.setTaskItemList(new ArrayList<>());
            taskInstanceInfo.setRebuildAssignTaskItemPlanCreate(false);
            return;
        }
    }
    //##<!-我是优美的功能模块分割线->##


    /**
     * 分片项只能是1-N,不能以0开头
     */
    public static List<TaskItem> createTaskItemList(WaitedAssignTaskInfo waitedAssignTaskInfo) {
        List<TaskItem> taskItemList = new ArrayList<>();
        String taskId = waitedAssignTaskInfo.getTaskId();
        String taskType = waitedAssignTaskInfo.getTaskType();
        String taskInstanceId = waitedAssignTaskInfo.getTaskInstanceId();
        String stage = waitedAssignTaskInfo.taskStage;
        String taskItemType = waitedAssignTaskInfo.getTaskStage$type();

        int totalTaskItemCount = waitedAssignTaskInfo.getTaskItemCount();
        int taskItemLoad = waitedAssignTaskInfo.getTaskItemLoad();
        Date nextFireDate = waitedAssignTaskInfo.getNextFireDate();
        Date timeoutDateTime = waitedAssignTaskInfo.getTimeoutDateTime();
        Date lockDateTime = waitedAssignTaskInfo.getLockDateTime();
        Date clearDateTime = waitedAssignTaskInfo.getClearDateTime();
        for (int taskItemIndex = 1; taskItemIndex <= totalTaskItemCount; taskItemIndex++) {
            TaskItem taskItem = new TaskItem();
            taskItem.setTaskType(taskType);
            taskItem.setTaskId(taskId);
            taskItem.setTaskInstanceId(taskInstanceId);
            taskItem.setTaskStage(stage);
            taskItem.setTaskStage$type(taskItemType);

            taskItem.setTaskStage$type$index(taskItemIndex);
            taskItem.setTaskItemLoad(taskItemLoad);
            taskItem.setNextFireDate(nextFireDate);
            taskItem.setTimeoutDateTime(timeoutDateTime);
            taskItem.setLockDateTime(lockDateTime);
            taskItem.setClearDateTime(clearDateTime);
            taskItemList.add(taskItem);
        }
        return taskItemList;
    }


    //##<!-我是优美的功能模块分割线->##

    /**
     * 更新最新的是否拥有流量和支持任务列表
     *
     * @param onlineExecutorList     old image
     * @param lastOnlineExecutorList 最新的
     */
    public static void updateIsOwnTrafficAndSupportedTaskIdList(List<Executor> onlineExecutorList, List<Executor> lastOnlineExecutorList) {
        for (Executor lastOnlineExecutor : lastOnlineExecutorList) {
            for (Executor onlineExecutor : onlineExecutorList) {
                if (lastOnlineExecutor.getExecutorId().equals(onlineExecutor.getExecutorId())) {
                    onlineExecutor.setNotFlowControl(lastOnlineExecutor.isNotFlowControl());
                    onlineExecutor.setTaskIdList(lastOnlineExecutor.getTaskIdList());
                }
            }
        }
    }

    /**
     * @param onlineExecutorList     old image
     * @param lastOnlineExecutorList 最新的
     */
    public static void addNewOnlineExecutorList(List<Executor> onlineExecutorList, List<Executor> lastOnlineExecutorList) {
        for (Executor lastOnlineExecutor : lastOnlineExecutorList) {
            boolean added = true;//是否需要添加(默认true)
            for (Executor onlineExecutor : onlineExecutorList) {
                if (onlineExecutor.getExecutorId().equals(lastOnlineExecutor)) {
                    added = false;
                    break;
                }
            }
            if (added) {
                onlineExecutorList.add(lastOnlineExecutor);
                continue;
            }
        }
    }

    /**
     * 找到并返回oldOnlineExecutorList(newOnlineExecutorList)中 enable/disable task id list,下线executor id/上线没有拥有流量的executor id list/上线拥有流量的executor id list
     */
    public static TaskItemAssignFixPoints getTaskItemAssignFixPoints(TaskItemAssignPresentSituation taskItemAssignPresentSituation) {
        //此处的oldOnlineExecutorList 可能下线 也可能处于上线失去流量也可能上线拥有流量的状态。
        List<Executor> oldOnlineExecutorList = taskItemAssignPresentSituation.oldOnlineExecutorList;
        //lastTaskIdList是没有任何意义的，只有lastEnableTaskIdList才进行任务分配
        List<String> lastTaskIdList = taskItemAssignPresentSituation.lastTaskIdList;
        //除开lastEnableTaskIdList都应该下线
        List<String> lastEnableTaskIdList = taskItemAssignPresentSituation.lastEnableTaskIdList;
        //
        List<String> lastOnlineExecutorIdList = taskItemAssignPresentSituation.lastOnlineExecutorIdList;
        List<String> lastOnlineOwnTrafficExecutorIdList = taskItemAssignPresentSituation.lastOnlineOwnTrafficExecutorIdList;
        List<String> lastOnlineDoNotOwnTrafficExecutorIdList = taskItemAssignPresentSituation.lastOnlineDoNotOwnTrafficExecutorIdList;
        //
        //收集oldOnlineExecutorList里面涉及的task id/executor id
        List<String> oldTaskIdList = new ArrayList<>();
        List<String> oldExecutorIdList = new ArrayList<String>();
        Set<String> oldTaskIdSet = new HashSet<>();
        Set<String> oldExecutorIdSet = new HashSet<String>();
        for (Executor onlineExecutor : oldOnlineExecutorList) {
            oldTaskIdSet.addAll(onlineExecutor.getTaskIdList());
            oldExecutorIdSet.add(onlineExecutor.getExecutorId());
        }
        oldTaskIdList.addAll(oldTaskIdSet);
        oldExecutorIdList.addAll(oldExecutorIdSet);
        //排除非交集部分，然后按照新的任务和执行器拓扑结构进行分配任务
        TaskItemAssignFixPoints taskItemAssignFixPoints = new TaskItemAssignFixPoints();
        taskItemAssignFixPoints.setTaskIdList(lastTaskIdList);
        {
            List<String> disableTaskIdList = new ArrayList<String>();
            oldTaskIdList.removeAll(lastEnableTaskIdList);
            disableTaskIdList.addAll(oldTaskIdList);
            taskItemAssignFixPoints.setDisableTaskIdList(disableTaskIdList);
        }
        taskItemAssignFixPoints.setEnableTaskIdList(lastEnableTaskIdList);
        //
        {
            List<String> offlineExecutorIdList = new ArrayList<String>();
            oldExecutorIdList.remove(lastOnlineExecutorIdList);
            offlineExecutorIdList.addAll(oldExecutorIdList);
            taskItemAssignFixPoints.setOfflineExecutorIdList(offlineExecutorIdList);
        }
        //
        taskItemAssignFixPoints.setOnlineExecutorIdList(lastOnlineExecutorIdList);
        taskItemAssignFixPoints.setOnlineOwnTrafficExecutorIdList(lastOnlineOwnTrafficExecutorIdList);
        taskItemAssignFixPoints.setOnlineDoNotOwnTrafficExecutorIdList(lastOnlineDoNotOwnTrafficExecutorIdList);
        //
        return taskItemAssignFixPoints;
    }

    //如果一个任务配置了优先list,那么需要把没有在优先配置列表的先移除,
    //如果没有配置则不管
    public static void fixPreferOnlineExecutorList(List<String> enableTaskIdList, PreferExecutorSettings preferExecutorSettings, List<TaskItem> pickUpEdTaskItemList, List<Executor> ownTrafficOnlineExecutorList) {
        for (String taskId : enableTaskIdList) {
            boolean preferExecutorIdListIsConfigured = preferExecutorSettings.taskIdToIsConfiguredPreferExecutorIdListMap.get(taskId); // 是否配置了preferList
            if (preferExecutorIdListIsConfigured) {
                List<String> preferExecutorIdListConfigured = preferExecutorSettings.taskIdToConfiguredPreferExecutorIdListMap.get(taskId); // 配置态的preferList
                if (preferExecutorIdListConfigured != null && preferExecutorIdListConfigured.size() > 0) {
                    for (Executor onlineExecutor : ownTrafficOnlineExecutorList) {
                        if (!preferExecutorIdListConfigured.contains(onlineExecutor.getExecutorId())) {
                            List<TaskItem> taskItemList = onlineExecutor.getTaskItemList();
                            Iterator<TaskItem> iterator = taskItemList.listIterator();
                            while (iterator.hasNext()) {
                                TaskItem taskItem = iterator.next();
                                if (taskItem.getTaskId().equals(taskId)) {
                                    onlineExecutor.setTotalLoadLevel(onlineExecutor.getTotalLoadLevel() - taskItem.getTaskItemLoad());
                                    iterator.remove();
                                    pickUpEdTaskItemList.add(taskItem);
                                    continue;
                                }
                            }
                        }
                    }
                }
            }
        }

    }

    public static List<Executor> getOnlineOwnTrafficButNotAssignedExecutorList(List<Executor> newOnlineExecutorList) {
        List<Executor> onlineOwnTrafficButNotAssignedExecutorList = newOnlineExecutorList.stream().filter(onlineExecutor -> onlineExecutor.isNotFlowControl() && onlineExecutor.getTaskItemList().isEmpty()).collect(Collectors.toList());
        return onlineOwnTrafficButNotAssignedExecutorList;
    }

    //##<!-我是优美的功能模块分割线->##
    public static void sortTaskItemList(List<TaskItem> taskItemList) {
        Collections.sort(taskItemList, new Comparator<TaskItem>() {//
            @Override
            public int compare(TaskItem o1, TaskItem o2) {
                int taskIdCompareResult = Integer.parseInt(o1.getTaskId()) - Integer.parseInt(o2.getTaskId());
                if (taskIdCompareResult != 0) {
                    if (taskIdCompareResult < 0) {
                        return -1;
                    } else {
                        return 1;
                    }
                }
                long taskInstanceIdCompareResult = Long.parseLong(o1.getTaskInstanceId()) - Long.parseLong(o2.getTaskInstanceId());
                if (taskInstanceIdCompareResult != 0) {
                    if (taskInstanceIdCompareResult < 0) {
                        return -1;
                    } else {
                        return 1;
                    }
                }
                int taskStageCompareResult = o1.getTaskStage().compareTo(o2.getTaskStage());
                if (taskStageCompareResult != 0) {
                    return taskStageCompareResult;
                }
                int taskItemTypeCompareResult = o1.getTaskStage$type().compareTo(o2.getTaskStage$type());
                if (taskItemTypeCompareResult != 0) {
                    return taskItemTypeCompareResult;
                }
                int taskItemLoadCompareResult = o1.getTaskItemLoad() - o2.getTaskItemLoad();
                if (taskItemLoadCompareResult < 0) {
                    return -1;
                } else if (taskItemLoadCompareResult == 0) {
                    return 0;
                } else {
                    return 1;
                }

            }
        });
    }
    //##<!-我是优美的功能模块分割线->##

    /**
     * preferExecutorIdList是否配置
     */
    public static boolean preferExecutorIdListIsConfigured(Curator curator, String taskId) throws Exception {
        if (curator.checkExists(TaskNodePaths.getTaskNodePath(taskId))) {
            String string = curator.getData(TaskNodePaths.getTaskBaseInfoNodePath(taskId, TaskNodePaths.prefer_executor_id_list));
            if (string != null) {
                return string.trim().length() > 0;
            }
        }
        return false;
    }

    /**
     * 获取当前配置且上线的preferExecutorIdSet
     */
    public static List<String> getPreferExecutorIdListConfigured(Curator curator, String taskId, List<String> executorIdList) throws Exception {
        Set<String> preferExecutorIdSet = new HashSet<>();
        if (curator.checkExists(TaskNodePaths.getTaskBaseInfoNodePath(taskId, TaskNodePaths.prefer_executor_id_list))) {
            String string = curator.getData(TaskNodePaths.getTaskBaseInfoNodePath(taskId, TaskNodePaths.prefer_executor_id_list));
            if (string != null) {
                String[] preferExecutorIds = string.split(",");
                for (String preferExecutorId : preferExecutorIds) {//preferExecutorIds的数据量相比executorIdList小
                    if (executorIdList.contains(preferExecutorId)) {
                        preferExecutorIdSet.add(preferExecutorId);
                    }
                }
            }
        }
        return new ArrayList<>(preferExecutorIdSet);
    }
    //##<!-我是优美的功能模块分割线->##

    public static void putBackWithPreferExecutorIdListConfigured(List<TaskItem> pickUpEdTaskItemList,//
                                                                 Map<String, List<Executor>> taskIdToOwnTrafficOnlineExecutorListMap,//
                                                                 Map<String, Boolean> taskIdToIsConfiguredPreferExecutorIdListMap,//
                                                                 Map<String, List<String>> taskIdToConfiguredPreferExecutorIdListMap) {//
        Iterator<TaskItem> iterator = pickUpEdTaskItemList.iterator();
        while (iterator.hasNext()) {
            TaskItem taskItem = iterator.next();
            String taskId = taskItem.getTaskId();
            // preferList为空不能作为判断是否配置preferList的依据,比如说配置了容器资源,但是全部下线了。
            if (taskIdToIsConfiguredPreferExecutorIdListMap.get(taskId)) {
                List<String> configuredPreferExecutorIdList = taskIdToConfiguredPreferExecutorIdListMap.get(taskId);
                List<Executor> preferOwnTrafficOnlineExecutorList = getPreferOwnTrafficOnlineExecutorList(taskIdToOwnTrafficOnlineExecutorListMap, taskId, configuredPreferExecutorIdList);
                // 如果存在preferExecutor，择优放回
                if (preferOwnTrafficOnlineExecutorList != null && !preferOwnTrafficOnlineExecutorList.isEmpty()) {
                    Executor onlineExecutor = getOnlineExecutorWithMinTaskTotalLoadLevelOnOnlineExecutor(preferOwnTrafficOnlineExecutorList, taskId);
                    addTaskItemToOnlineExecutor(taskItem, onlineExecutor);
                    iterator.remove();
                } else {
                    // 如果不存在preferExecutor 等到后续进行放回操作，避免不均衡的情况
                }
            }
        }
    }

    public static List<Executor> getPreferOwnTrafficOnlineExecutorList(Map<String, List<Executor>> taskIdToOwnTrafficOnlineExecutorListMap, String taskId, List<String> configuredPreferExecutorIdList) {
        List<Executor> preferOnlineExecutorList = new ArrayList<>();
        List<Executor> ownTrafficOnlineExecutorList = taskIdToOwnTrafficOnlineExecutorListMap.get(taskId);
        for (int i = 0; i < ownTrafficOnlineExecutorList.size(); i++) {
            Executor onlineExecutor = ownTrafficOnlineExecutorList.get(i);
            if (configuredPreferExecutorIdList.contains(onlineExecutor.getExecutorId())) {
                preferOnlineExecutorList.add(onlineExecutor);
            }
        }
        return preferOnlineExecutorList;
    }

    public static void putBackWithoutPreferExecutorIdListConfigured(List<TaskItem> pickUpEdTaskItemList, Map<String, List<Executor>> taskIdToOwnTrafficOnlineExecutorListMap) {
        Iterator<TaskItem> iterator = pickUpEdTaskItemList.iterator();
        while (iterator.hasNext()) {
            TaskItem taskItem = iterator.next();
            List<Executor> taskOwnTrafficOnlineExecutorList = taskIdToOwnTrafficOnlineExecutorListMap.get(taskItem.getTaskId());//重点是这段代码 如果没有执行器怎么办???
            if (taskOwnTrafficOnlineExecutorList != null && !taskOwnTrafficOnlineExecutorList.isEmpty()) {
                Executor onlineExecutor = getOnlineExecutorWithMinTaskTotalLoadLevelOnOnlineExecutor(taskOwnTrafficOnlineExecutorList, taskItem.getTaskId());
                addTaskItemToOnlineExecutor(taskItem, onlineExecutor);
                iterator.remove();
            }

        }
    }

    /**
     * 获取该任务负荷最小的executor，如果相同，那么取所有任务负荷最小的executor
     */
    public static Executor getOnlineExecutorWithMinTaskTotalLoadLevelOnOnlineExecutor(List<Executor> taskOwnTrafficOnlineExecutorList, String taskId) {
        Executor minTaskTotalLoadLevelOnlineExecutor = null;
        int minTaskTotalLoadLevelOnOnlineExecutor = 0;
        for (int i = 0; i < taskOwnTrafficOnlineExecutorList.size(); i++) {
            Executor onlineExecutor = taskOwnTrafficOnlineExecutorList.get(i);
            int taskTotalLoadLevelOnExecutor = getTaskTotalLoadLevelOnExecutor(onlineExecutor, taskId);
            if (minTaskTotalLoadLevelOnlineExecutor == null ||//
                    minTaskTotalLoadLevelOnOnlineExecutor > taskTotalLoadLevelOnExecutor//
                    || minTaskTotalLoadLevelOnOnlineExecutor == taskTotalLoadLevelOnExecutor && minTaskTotalLoadLevelOnlineExecutor.getTotalLoadLevel() > onlineExecutor.getTotalLoadLevel()//
            ) {
                minTaskTotalLoadLevelOnlineExecutor = onlineExecutor;
                minTaskTotalLoadLevelOnOnlineExecutor = taskTotalLoadLevelOnExecutor;
            }
        }
        return minTaskTotalLoadLevelOnlineExecutor;
    }

    /**
     * 计算指定任务在特定executor上的负荷
     */
    public static int getTaskTotalLoadLevelOnExecutor(Executor executor, String taskId) {
        int taskTotalLoadLevel = 0;
        List<TaskItem> taskItemList = executor.getTaskItemList();
        for (int i = 0; i < taskItemList.size(); i++) {
            TaskItem taskItem = taskItemList.get(i);
            if (taskId.equals(taskItem.getTaskId())) {
                taskTotalLoadLevel += taskItem.getTaskItemLoad();
            }
        }
        return taskTotalLoadLevel;
    }

    public static void addTaskItemToOnlineExecutor(TaskItem taskItem, Executor onlineExecutor) {
        if (onlineExecutor != null) {
            if (isIn(onlineExecutor.getTaskItemList(), taskItem)) {
                log.error("The shard({}-{}) is running in the executor of {}, cannot be put again", taskItem.getTaskId(), taskItem.getTaskStage$type$index(), onlineExecutor.getExecutorId());
            } else {
                onlineExecutor.getTaskItemList().add(taskItem);
                onlineExecutor.setTotalLoadLevel(onlineExecutor.getTotalLoadLevel() + taskItem.getTaskItemLoad());
            }
        } else {
            log.info("No executor to take over the shard: {}-{}", taskItem.getTaskId(), taskItem.getTaskStage$type$index());
        }
    }

    public static boolean isIn(List<TaskItem> taskItemList, TaskItem taskItem) {
        for (int i = 0; i < taskItemList.size(); i++) {
            TaskItem $taskItem = taskItemList.get(i);
            if ($taskItem.getTaskId().equals(taskItem.getTaskId()) &&//
                    $taskItem.getTaskInstanceId().equals(taskItem.getTaskInstanceId()) &&//

                    $taskItem.getTaskStage$type$index() == taskItem.getTaskStage$type$index()) {//
                return true;
            }
        }
        return false;
    }

    //##<!-我是优美的功能模块分割线->##


    /**
     * 刷新onlineExecutorList taskIdList
     */
    public static boolean refreshOnlineExecutorSupportedTaskIdList(Curator curator, List<Executor> onlineExecutorList, String taskId) throws Exception {
        boolean fixed = false;
        for (int i = 0; i < onlineExecutorList.size(); i++) {
            Executor onlineExecutor = onlineExecutorList.get(i);
            if (onlineExecutor.getTaskIdList() == null) {
                onlineExecutor.setTaskIdList(new ArrayList<String>());
            }

            String executorTasksNodePath = ExecutorTaskNodePath.getExecutorTasksNodePath(onlineExecutor.getExecutorId());
            if (curator.checkExists(executorTasksNodePath)) {
                onlineExecutor.getTaskIdList().clear();
                onlineExecutor.getTaskIdList().addAll(curator.getChildren(executorTasksNodePath));
            } else {
                onlineExecutor.getTaskIdList().clear();
            }
        }
        return fixed;
    }

    /**
     * 移除taskId对应的所有的Task Item List
     */
    public static List<TaskItem> removeTaskItemListOfTask(List<Executor> ownTrafficOnlineExecutorList, String taskId) {
        List<TaskItem> removedTaskItemList = new ArrayList<>();
        for (int i = 0; i < ownTrafficOnlineExecutorList.size(); i++) {
            Executor onlineExecutor = ownTrafficOnlineExecutorList.get(i);
            Iterator<TaskItem> iterator = onlineExecutor.getTaskItemList().iterator();
            while (iterator.hasNext()) {
                TaskItem taskItem = iterator.next();
                if (taskId.equals(taskItem.getTaskId())) {//包括该任务下的所有的任务实例
                    onlineExecutor.setTotalLoadLevel(onlineExecutor.getTotalLoadLevel() - taskItem.getTaskItemLoad());
                    iterator.remove();
                    removedTaskItemList.add(taskItem);
                }
            }
        }
        return removedTaskItemList;
    }


    public static boolean hasTaskItemRunning(String taskId, List<Executor> onlineOwnTrafficExecutorList) {
        for (int i = 0; i < onlineOwnTrafficExecutorList.size(); i++) {
            List<TaskItem> taskItemList = onlineOwnTrafficExecutorList.get(i).getTaskItemList();
            for (int j = 0; j < taskItemList.size(); j++) {
                if (taskItemList.get(j).getTaskId().equals(taskId)) {
                    return true;
                }
            }
        }
        return false;
    }

    public static List<Executor> getOnlineExecutorListInPreferOrDisPreferExecutorIdListConfigured(
            String taskId,//
            List<String> preferExecutorIdListConfigured,//
            List<Executor> onlineOwnTrafficExecutorList) {//
        List<Executor> onlineExecutorList = new ArrayList<>();
        for (Executor onlineExecutor : onlineOwnTrafficExecutorList) {
            if (onlineExecutor.getTaskIdList().contains(taskId) && preferExecutorIdListConfigured.contains(onlineExecutor.getExecutorId())) {
                onlineExecutorList.add(onlineExecutor);
            }
        }
        return onlineExecutorList;
    }

    /**
     * 平衡摘取算法-非多余权重保留模式。保证单个任务均衡不影响所有执行节点。
     *
     * @param onlineExecutorList
     * @param taskItemList
     */
    public static void balancePickUp(List<Executor> onlineExecutorList, String taskId, List<TaskItem> taskItemList) {
        // 计算出executor对该任务的平均负荷
        int totalLoadLevel = 0;
        for (TaskItem taskItem : taskItemList) {
            totalLoadLevel += taskItem.getTaskItemLoad();
        }
        for (Executor onlineExecutor : onlineExecutorList) {
            for (TaskItem taskItem : onlineExecutor.getTaskItemList()) {
                if (taskId.equals(taskItem.getTaskId())) {
                    totalLoadLevel += taskItem.getTaskItemLoad();
                }
            }
        }
        int averageTotalLoadLevelAtPerExecutor = totalLoadLevel / onlineExecutorList.size();
        //
        // 摘取executor超出平均负荷的分片
        for (Executor onlineExecutor : onlineExecutorList) {
            int taskTotalLoadLevel = 0;
            Iterator<TaskItem> iterator = onlineExecutor.getTaskItemList().iterator();
            while (iterator.hasNext()) {
                TaskItem taskItem = iterator.next();
                if (taskId.equals(taskItem.getTaskId())) {
                    //taskTotalLoadLevel + taskItem.getLoadLevel()超过平均权重则不保留。
                    if (averageTotalLoadLevelAtPerExecutor == 0/**很细微的点  可能平均权重为0*/ || taskTotalLoadLevel + taskItem.getTaskItemLoad() > averageTotalLoadLevelAtPerExecutor) {
                        iterator.remove();
                        onlineExecutor.setTotalLoadLevel(onlineExecutor.getTotalLoadLevel() - taskItem.getTaskItemLoad());
                        //
                        taskItemList.add(taskItem);
                    } else {
                        //不超过平均权重则保留。
                        taskTotalLoadLevel += taskItem.getTaskItemLoad();
                    }
                }
            }
        }
    }

    /**
     * getSupportedTaskIdOwnTrafficOnlineExecutorList
     */
    public static List<Executor> getSupportedTaskIdOwnTrafficOnlineExecutorList(List<Executor> onlineOwnTrafficExecutorList, String taskId) {
        List<Executor> list = new ArrayList<>();
        for (Executor onlineExecutor : onlineOwnTrafficExecutorList) {
            List<String> taskIdList = onlineExecutor.getTaskIdList();
            if (onlineExecutor.isNotFlowControl() && taskIdList != null && taskIdList.contains(taskId)) {
                list.add(onlineExecutor);
            }
        }
        return list;
    }


    public static List<TaskInstanceInfo> getSimpleTaskStateFullHasAssignedTaskInstanceInfoList(ScheduleManager scheduleManager) {
        List<TaskInstanceInfo> taskInstanceInfoList = new ArrayList<>();
        for (List<TaskInstanceInfo> list : scheduleManager.stateFullSimpleTaskTaskInstanceManager.assignedTasks.values()) {
            for (TaskInstanceInfo taskInstanceInfo : list) {
                if (!taskInstanceInfo.isCancelled()) {
                    taskInstanceInfoList.add(taskInstanceInfo);//如果当前此次调度没有取消，那么下次进行取消
                }
            }
        }
        return taskInstanceInfoList;
    }

    public static List<TaskInstanceInfo> getSimpleTaskStateLessHasAssignedTaskInstanceInfoList(ScheduleManager scheduleManager) {

        List<TaskInstanceInfo> taskInstanceInfoList = new ArrayList<>();
        for (List<TaskInstanceInfo> list : scheduleManager.stateLessSimpleTaskTaskInstanceManager.assignedTasks.values()) {
            for (TaskInstanceInfo taskInstanceInfo : list) {
                if (!taskInstanceInfo.isCancelled()) {
                    taskInstanceInfoList.add(taskInstanceInfo);//如果当前此次调度没有取消，那么下次进行取消
                }
            }
        }
        return taskInstanceInfoList;
    }

    public static List<TaskInstanceInfo> getBatchTaskHasAssignedTaskInstanceInfoList(ScheduleManager scheduleManager) {
        List<TaskInstanceInfo> taskInstanceInfoList = new ArrayList<>();
        for (List<TaskInstanceInfo> list : scheduleManager.batchTaskTaskInstanceManager.assignedTasks.values()) {
            for (TaskInstanceInfo taskInstanceInfo : list) {
                if (!taskInstanceInfo.isCancelled()) {
                    taskInstanceInfoList.add(taskInstanceInfo);//如果当前此次调度没有取消，那么下次进行取消
                }
            }
        }
        return taskInstanceInfoList;
    }


}
